package gupao.concurrency.lockfree.disruptor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import gupao.concurrency.utils.ThreadPoolService;
import gupao.concurrency.utils.Utils;
import lombok.extern.slf4j.Slf4j;

import java.util.Random;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public class DisruptorSample {

    static final int PRODUCER_CONCURRENCY = 5;
    static final int PRODUCT_NUM_EACH = 5;
    static ThreadPoolExecutor PRODUCER_THREAD_POOL = ThreadPoolService.getOrCreateThreadPoolExecutor("producer",
            PRODUCER_CONCURRENCY,PRODUCER_CONCURRENCY);

    public static void main(String[] args) {
//        Disruptor<OrderEvent> disruptor = createDisruptor(ProducerType.SINGLE, new YieldingWaitStrategy());
        Disruptor<OrderEvent> disruptor = createDisruptor(ProducerType.MULTI, new YieldingWaitStrategy());
        disruptor.handleEventsWith(createConsumers(1));
        disruptor.start();
        createAndStartProducer(disruptor);
        Utils.sleepIgnoreExp(60000);
    }

    static void createAndStartProducer(Disruptor<OrderEvent> disruptor){
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
        OrderEventProducer eventProducer = new OrderEventProducer(ringBuffer);
        for(int i = 0; i < PRODUCER_CONCURRENCY; i++){
            PRODUCER_THREAD_POOL.submit(() -> {
                log.info("{} is start running",Thread.currentThread().getName());
                for(int k = 0; k < PRODUCT_NUM_EACH; k++){
                    Utils.sleepIgnoreExp(1000);
                    String order = UUID.randomUUID().toString();
                    log.info("{} produce order: {}", Thread.currentThread().getName(), order);
                    eventProducer.onData(order);
                }
                log.info("{} done.",Thread.currentThread().getName());
            });
        }
    }

    static DefaultOrderEventHandler[] createConsumers(int num){
        int x = Math.max(1, num);
        DefaultOrderEventHandler[] results = new DefaultOrderEventHandler[x];
        for(int i = 0; i < x; i++){
            results[i] = new DefaultOrderEventHandler();
        }
        return results;
    }

    static Disruptor<OrderEvent> createDisruptor(ProducerType type, WaitStrategy strategy){
        return new Disruptor<>(
                OrderEvent::new,
                1024 * 1024,
                Executors.defaultThreadFactory(),
                type,
                strategy
        );
    }

}
